package spring.cloud.tasks.api.service.impl;

import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.api.domain.task.TaskConfiguration;
import spring.cloud.tasks.api.domain.task.TaskType;
import spring.cloud.tasks.common.node.TaskConfigurationNode;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.TransactionalZooKeeper;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;

import java.util.Map;

@Slf4j
public final class TaskConfigurationAndZooKeeperUtil {
    public static TaskConfiguration constructJobConfigOfCreate(String nameSpace, TaskConfiguration taskConfiguration, Map<String, Object> reqParams)
            throws RuntimeException {

        String taskType = (String) reqParams.get("taskType");
        if (TaskType.UNKNOWN_TASK == TaskType.getTaskType(taskType)) {
            throw new RuntimeException(String.format("taskType", "is malformed"));
        }
        taskConfiguration.setTaskType(taskType);
        taskConfiguration.setTaskId((String) reqParams.get("taskId"));
        taskConfiguration.setXId((String) reqParams.get("xId"));
        taskConfiguration.setDescription((String) reqParams.get("description"));
        //
        taskConfiguration.setCronExpression((String) reqParams.get("cronExpression"));
        taskConfiguration.setPausePeriodDate((String) reqParams.get("pausePeriodDate"));
        taskConfiguration.setPausePeriodTime((String) reqParams.get("pausePeriodTime"));
        //
        taskConfiguration.setTimeoutSeconds((Integer) reqParams.get("timeoutSeconds"));
        taskConfiguration.setTimeout4AlarmSeconds((Integer) reqParams.get("timeout4AlarmSeconds"));
        //
        taskConfiguration.setTaskItemTotalCount((Integer) reqParams.get("taskItemTotalCount"));
        taskConfiguration.setLoadLevel((Integer) reqParams.get("loadLevel"));
        //
        taskConfiguration.setTaskParameter((String) reqParams.get("taskParameter"));
        taskConfiguration.setTaskItemParameters((String) reqParams.get("taskItemParameters"));
        taskConfiguration.setTaskCustomContext((String) reqParams.get("taskCustomContext"));
        //
        taskConfiguration.setEnabled((String) reqParams.get("enabled"));
        taskConfiguration.setPreferExecutorIdList((String) reqParams.get("preferExecutorIdList"));
        return taskConfiguration;
    }

    public static void createTaskConfigurationToZooKeeper(ZooKeeper zooKeeper, TaskConfiguration taskConfiguration) throws RuntimeException {
        try {
            String taskId = taskConfiguration.getTaskId();
            // 添加任务根节点和config结点
            zooKeeper.create(TaskNodePath.getTaskConfigurationNodePath(taskId), "");
            TransactionalZooKeeper transactionalZooKeeper = zooKeeper.useTransaction();
            // 添加任务
            transactionalZooKeeper
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_type), taskConfiguration.getTaskType())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_id), taskConfiguration.getTaskId())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.xid), taskConfiguration.getXId())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.description), taskConfiguration.getDescription())
                    //
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.cron_expression), taskConfiguration.getCronExpression())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_date), taskConfiguration.getPausePeriodDate())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_time), taskConfiguration.getPausePeriodTime())
                    //
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_seconds), taskConfiguration.getTimeoutSeconds())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_4_alarm_seconds), taskConfiguration.getTimeout4AlarmSeconds())
                    //
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_total_count), taskConfiguration.getTaskItemTotalCount())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.load_level), taskConfiguration.getLoadLevel())
                    //
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_parameters), taskConfiguration.getTaskItemParameters())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_parameter), taskConfiguration.getTaskParameter())
                    //
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled), taskConfiguration.getEnabled())
                    .create(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list), taskConfiguration.getPreferExecutorIdList());
            //
            transactionalZooKeeper.commit();
        } catch (Exception e) {
            log.error("create job to zk failed", e);
            throw new RuntimeException(e);
        }
    }

    public static void updateTaskConfigurationToZooKeeper(ZooKeeper zooKeeper, TaskConfiguration taskConfiguration) throws RuntimeException {
        try {
            String taskId = taskConfiguration.getTaskId();

            TransactionalZooKeeper transactionalZooKeeper = zooKeeper.useTransaction();

            // 更新任务
            transactionalZooKeeper

                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_type), taskConfiguration.getTaskType())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_id), taskConfiguration.getTaskId())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.xid), taskConfiguration.getXId())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.description), taskConfiguration.getDescription())
                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.cron_expression), taskConfiguration.getCronExpression())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_date), taskConfiguration.getPausePeriodDate())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_time), taskConfiguration.getPausePeriodTime())
                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_seconds), taskConfiguration.getTimeoutSeconds())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_4_alarm_seconds), taskConfiguration.getTimeout4AlarmSeconds())

                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_total_count), taskConfiguration.getTaskItemTotalCount())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.load_level), taskConfiguration.getLoadLevel())
                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_parameters), taskConfiguration.getTaskItemParameters())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_parameter), taskConfiguration.getTaskParameter())
                    //
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled), taskConfiguration.getEnabled())
                    .replaceIfChanged(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list), taskConfiguration.getPreferExecutorIdList());
            //

            transactionalZooKeeper.commit();
        } catch (Exception e) {
            log.error("update job to zk failed", e);
            throw new RuntimeException(e);
        }
    }


    public static TaskConfiguration getTaskConfigurationFromZooKeeper(ZooKeeper zooKeeper, String taskId) throws RuntimeException {
        String configNodePath = TaskNodePath.getTaskConfigurationNodePath(taskId);
        if (!zooKeeper.checkExists(configNodePath)) {
            return null;//
        }
        //ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
        TaskConfiguration taskConfiguration = new TaskConfiguration();
        taskConfiguration.setTaskType(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_type)));
        taskConfiguration.setTaskId(taskId);
        taskConfiguration.setXId(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.xid)));
        taskConfiguration.setDescription(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.description)));
        //
        taskConfiguration.setCronExpression(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.cron_expression)));
        taskConfiguration.setPausePeriodDate(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_date)));
        taskConfiguration.setPausePeriodTime(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.pause_period_time)));
        //
        taskConfiguration.setTimeoutSeconds(Integer.valueOf(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_seconds))));
        taskConfiguration.setTimeout4AlarmSeconds(Integer.valueOf(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.timeout_4_alarm_seconds))));
        //
        taskConfiguration.setTaskItemTotalCount(Integer.valueOf(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_total_count))));
        taskConfiguration.setLoadLevel(Integer.valueOf(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.load_level))));
        //
        taskConfiguration.setTaskItemParameters(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_item_parameters)));
        taskConfiguration.setTaskParameter(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskConfigurationNode.task_parameter)));
        //
        taskConfiguration.setEnabled(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.enabled)));// 默认是禁用的
        taskConfiguration.setPreferExecutorIdList(zooKeeper.getData(TaskNodePath.getTaskConfigurationNodePath(taskId, TaskNodePath.prefer_executor_id_list)));
        //

        return taskConfiguration;
    }
}
